package qq2564874169.tuaotuao.fx.caching.zookeeper;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import qq2564874169.tuaotuao.fx.FxAssert;
import qq2564874169.tuaotuao.fx.FxDate;
import qq2564874169.tuaotuao.fx.Var;
import qq2564874169.tuaotuao.fx.caching.RemoteCache;
import qq2564874169.tuaotuao.fx.caching.RemoteCacheItem;
import qq2564874169.tuaotuao.fx.utils.ExUtil;

import java.nio.charset.StandardCharsets;

public class ZkCache extends RemoteCache {
    private static String rootNode;
    private static final String sliExpNode = "get_time";
    private static final String dataNode = "data_";
    private static final int dataMaxSize = 1024 * 512;
    private ZooKeeper zk;


    public static void createRootNode(ZooKeeper zk) {
        createRootNode(zk, "/FxCache");
    }

    public static void createRootNode(ZooKeeper zk, String nodeName) {
        try {
            if (zk.exists(nodeName, false) == null) {
                zk.create(nodeName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            rootNode = nodeName;
        } catch (KeeperException | InterruptedException e) {
            throw ExUtil.toRuntime(e);
        }
    }

    public ZkCache(ZooKeeper zk) {
        this.zk = zk;
    }

    private String fullPath(String path) {
        FxAssert.isNotNull(rootNode, "请先调用createRootNode创建根节点");
        if (path.startsWith("/")) {
            return rootNode + path;
        } else {
            return rootNode + "/" + path;
        }
    }

    private void deleteNode(String key) {
        try {
            var list = zk.getChildren(key, false);
            for (var k : list) {
                deleteNode(key + "/" + k);
            }
            zk.delete(key, -1);
        } catch (KeeperException | InterruptedException e) {
            throw ExUtil.toRuntime(e);
        }
    }

    private RemoteCacheItem existsNode(String path) {
        try {
            var gs = new Stat();
            var data = zk.getData(path, false, gs);
            var str = new String(data, StandardCharsets.UTF_8);
            var item = (RemoteCacheItem) valueDeserialize(str, RemoteCacheItem.class);

            if (item.absoluteExpiration != null) {
                var exp = new FxDate(gs.getCtime()).addSecond(item.absoluteExpiration);
                if (exp.yyyyMMddHHmmss() <= FxDate.currentTime()) {
                    deleteNode(path);
                    item = null;
                }
            }
            if (item != null && item.slidingExpiration != null) {
                var tnode = path + "/" + sliExpNode;
                var tdata = zk.getData(tnode, false, new Stat());
                var tstr = new String(tdata, StandardCharsets.UTF_8);
                var time = FxDate.tryParse(tstr);
                if (time.addSecond(item.slidingExpiration).yyyyMMddHHmmss() <= FxDate.currentTime()) {
                    deleteNode(path);
                    item = null;
                }
            }
            return item;
        } catch (InterruptedException e) {
            throw ExUtil.toRuntime(e);
        } catch (KeeperException e) {
            if (e.code().equals(KeeperException.Code.NONODE)) {
                return null;
            }
            throw ExUtil.toRuntime(e);
        }
    }

    @Override
    protected void saveRemote(RemoteCacheItem item) {
        var path = fullPath(item.key);
        var stat = new Var<>(existsNode(path));
        try {
            if (!stat.isNull()) {
                deleteNode(path);
            }
            var seria = valueSerialize(item).getBytes(StandardCharsets.UTF_8);
            if (item.slidingExpiration != null) {
                zk.create(path, seria, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                zk.create(path + "/" + sliExpNode, new FxDate().yyyyMMddHHmmss().toString().getBytes(StandardCharsets.UTF_8),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } else {
                zk.create(path, seria, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (KeeperException | InterruptedException e) {
            throw ExUtil.toRuntime(e);
        }
    }

    @Override
    protected RemoteCacheItem getRemoteItem(String key) {
        try {
            var path = fullPath(key);
            var stat = existsNode(path);
            if (stat != null) {
                var data = zk.getData(path, false, new Stat());
                var str = new String(data, StandardCharsets.UTF_8);
                var item = (RemoteCacheItem) valueDeserialize(str, RemoteCacheItem.class);
                if (item.slidingExpiration != null) {
                    var tnode = path + "/" + sliExpNode;
                    zk.setData(tnode, new FxDate().yyyyMMddHHmmss().toString().getBytes(StandardCharsets.UTF_8), -1);
                }
                return item;
            }
            return null;
        } catch (Exception e) {
            throw ExUtil.toRuntime(e);
        }
    }

    @Override
    protected void deleteRemote(String key) {
        var p = fullPath(key);
        var stat = existsNode(p);
        if (stat != null) {
            deleteNode(p);
        }
    }

    @Override
    protected boolean existsRemote(String key) {
        return existsNode(fullPath(key)) != null;
    }
}
